package com.atbeijing.bigdata.spark.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark03_Req_PageFlow_1 {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
        val sc = new SparkContext(conf)

        val action = sc.textFile("data/user_visit_action.txt")

        val map : Map[String, Int] = Map(
            ("a", 1)
        )
        map.toList
        val list : List[(String, Int)] = List(
            ("a", 1)
        )


        val actionDatas = action.map(
            line => {
                val datas = line.split("_")
                UserVisitAction(
                    datas(0),
                    datas(1).toLong,
                    datas(2),
                    datas(3).toLong,
                    datas(4),
                    datas(5),
                    datas(6).toLong,
                    datas(7).toLong,
                    datas(8),
                    datas(9),
                    datas(10),
                    datas(11),
                    datas(12).toLong
                )
            }
        )

        // [1,2,3,4,5,6,7]
        // [1-2,2-3,3-4,4-5,5-6, 6-7]
        val allowIds = List(1L,2L,3L,4L,5L,6L,7L)
        val zipAllowIds: List[(Long, Long)] = allowIds.zip(allowIds.tail)

        // TODO 统计分母 (页面ID, Count)
        val pageCountMap: Map[Long, Int] = actionDatas.filter(
            action => {
                allowIds.init.contains(action.page_id.toInt)
            }
        ).map(
            action => {
                (action.page_id, 1)
            }
        ).reduceByKey(_ + _).collect.toMap

        // 统计分子
        // 将数据按照session进行分组
        actionDatas.cache()
        val sessionData: RDD[(String, Iterable[UserVisitAction])] = actionDatas.groupBy(_.session_id)

        // 将分组后的数据根据页面的访问时间进行排序(升序)
        // (String, List[(String, Int)])
        val mapDatas: RDD[(String, List[(String, Int)])] = sessionData.mapValues(
            iter => {
                val actions: List[UserVisitAction] = iter.toList.sortBy(_.action_time)
                // 【1，8, 2，9, 3，10,4，5】
                // 【2，3，4, 5】
                val ids: List[Long] = actions.map(_.page_id)
//                // 【1, 2， 3,4，5】
//                ids.filter(
//                    id => {
//                        allowIds.init.contains(id)
//                    }
//                )
                // 【1-8, 8-2，2-9, 9-3】
                // 【1-2，2-3，3-4，4-5】
                val flowids: List[(Long, Long)] = ids.zip(ids.tail)

                flowids.filter(
                    t => {
                        zipAllowIds.contains(t)
                    }
                ).map {
                    case (id1, id2) => {
                        (id1 + "-" + id2, 1)
                    }
                }
            }
        )

        // RDD[(String, Int)] => ("a", 1), ("a", 2)
        // RDD[List[(String, Int)]] => List( ("a",1) ), List(("a", 2))
        val flatDatas: RDD[(String, Int)] = mapDatas.map(_._2).flatMap(list=>list)
        val pageFlowSum: RDD[(String, Int)] = flatDatas.reduceByKey(_ + _)

        // TODO 计算页面单挑转换率
        // 分子 / 分母
        pageFlowSum.foreach {
            case ( pageflow, sum ) => {
                val ids = pageflow.split("-")
                val cnt = pageCountMap.getOrElse(ids(0).toLong, 1)
                println(s"页面跳转【${pageflow}】的转换率为 = " + (sum.toDouble/cnt))
            }
        }

        sc.stop()

    }
    case class UserVisitAction(
          date: String,//用户点击行为的日期
          user_id: Long,//用户的ID
          session_id: String,//Session的ID
          page_id: Long,//某个页面的ID
          action_time: String,//动作的时间点
          search_keyword: String,//用户搜索的关键词
          click_category_id: Long,//某一个商品品类的ID
          click_product_id: Long,//某一个商品的ID
          order_category_ids: String,//一次订单中所有品类的ID集合
          order_product_ids: String,//一次订单中所有商品的ID集合
          pay_category_ids: String,//一次支付中所有品类的ID集合
          pay_product_ids: String,//一次支付中所有商品的ID集合
          city_id: Long//城市 id
      )
}
